package com.t2t.top.framework.rocketmq.client.extension.impl;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.t2t.top.framework.rocketmq.client.extension.MessageProducer;
import com.t2t.top.framework.rocketmq.client.extension.SerializeType;
import com.t2t.top.framework.rocketmq.utils.DiamondUtil;
import com.t2t.top.framework.rocketmq.utils.RunTimeUtil;
import com.taobao.diamond.manager.ManagerListenerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.Serializable;

public class MessageProducerImpl implements InitializingBean, DisposableBean, MessageProducer {

    Logger logger = LoggerFactory.getLogger(getClass());

    private SerializeType serializeType = SerializeType.FASTJSON;

    //begin:rocketmq生产者配置######################################
    private int maxMessageSize = 4 * 1024 * 1024;//最大可发4M消息
    private String topic;// 必需配置： 如, order
    private String tags;// 必需配置：tags,如：TagA
    private DefaultMQProducer producer;
    //end:rocketmq生产者配置######################################

    private String dataId;// diamond
    private String group;// diamond

    @Override
    public void destroy() throws Exception {
        producer.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        registDiamondListener();
    }

    private void registDiamondListener() {
        DiamondUtil.registDiamondListener(new ManagerListenerAdapter() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                restartProducer(configInfo);
            }
        }, group, dataId);
    }

    public synchronized void restartProducer(String configInfo) {
        DefaultMQProducer producerNew = null;
        try {
            producerNew = new DefaultMQProducer(topic);
            producerNew.setNamesrvAddr(configInfo);
            producerNew.setMaxMessageSize(maxMessageSize);
            producerNew.setInstanceName(RunTimeUtil.getRocketMqUniqeInstanceName());
            producerNew.start();
        } catch (MQClientException e) {
            logger.error("rocketmq restartProducer error ", e);
            return;
        }
        /** 这里很重要：先替换老实例再关闭老实例;如果先关闭老实例,有可能它在被程序调用着 */
        DefaultMQProducer oldInsetance = producer;
        producer = producerNew;
        if (oldInsetance != null) {
            oldInsetance.shutdown();
        }
    }

    @Override
    public <E extends Serializable> SendResult send(String key, E messageObject) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // if (key == null || "".equals(key)) {
        // throw new IllegalArgumentException("rocketmq message without key");
        // }
        //SerializerFeature.BrowserCompatible 兼容手机端或者其他浏览器的特殊符号
        Message msg = new com.alibaba.rocketmq.common.message.Message(
                topic,// topic
                tags,// tag
                key,// keys
                //JSON.toJSONBytes(messageObject, SerializerFeature.WriteClassName, SerializerFeature.BrowserCompatible)// body
                serializeType.serialize(messageObject)
        );

        SendResult sendResult = null;
        sendResult = producer.send(msg);
        logger.debug("rocketmq send success,key=" + key + ";sendResult:" + sendResult);
        return sendResult;

    }

    public SendResult send(String key, byte[] message) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // if (key == null || "".equals(key)) {
        // throw new IllegalArgumentException("rocketmq message without key");
        // }
        //SerializerFeature.BrowserCompatible 兼容手机端或者其他浏览器的特殊符号
        Message msg = new com.alibaba.rocketmq.common.message.Message(
                topic,// topic
                tags,// tag
                key,// keys
                //JSON.toJSONBytes(messageObject, SerializerFeature.WriteClassName, SerializerFeature.BrowserCompatible)// body
                message
        );

        SendResult sendResult = null;
        sendResult = producer.send(msg);
        logger.debug("rocketmq send success,key=" + key + ";sendResult:" + sendResult);
        return sendResult;
    }

    public <E extends Serializable> void send(String key, E messageObject, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
        // if (key == null || "".equals(key)) {
        // throw new IllegalArgumentException("rocketmq message without key");
        // }
        //SerializerFeature.BrowserCompatible 兼容手机端或者其他浏览器的特殊符号
        Message msg = new com.alibaba.rocketmq.common.message.Message(
                topic,// topic
                tags,// tag
                key,// keys
                //JSON.toJSONBytes(messageObject, SerializerFeature.WriteClassName, SerializerFeature.BrowserCompatible)// body
                serializeType.serialize(messageObject)
        );

        producer.send(msg, sendCallback);
        logger.debug("rocketmq send key=" + key);
    }

    public int getMaxMessageSize() {
        return maxMessageSize;
    }

    public void setMaxMessageSize(int maxMessageSize) {
        this.maxMessageSize = maxMessageSize;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    public void setDataId(String dataId) {
        this.dataId = dataId;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public void setSerializeType(String serializeType) {
        this.serializeType = SerializeType.valueOf(serializeType != null ? serializeType.toUpperCase() : "FASTJSON");
    }

    public SerializeType getSerializeType() {
        return serializeType;
    }

    public void setSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
    }

    public MQProducer getMQProducer() {
        return producer;
    }
}